package com.chukun.flink.stream.operator.base;

import com.chukun.flink.stream.bean.Trade;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

/**
 * @author chukun
 * @version 1.0.0
 * @description 聚合类型的操作符
 * @createTime 2022年05月10日 00:06:00
 */
public class AggregationsOperator {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Trade> streamSource = env.fromCollection(new ArrayList<Trade>(){
            {
                add(new Trade("12300986754",899,"2018-06"));
                add(new Trade("12300986754",699,"2018-07"));
                add(new Trade("18800986754",88,"2018-08"));
                add(new Trade("18800986754",99,"2018-09"));
            }
        });

        KeyedStream<Trade, String> resultStream = streamSource.keyBy((KeySelector<Trade, String>) Trade::getPhone);

        // sum操作
        resultStream.sum("amount").print("sum");


        // min操作，12300986754这行最小的数据，输出的时间还是  2018-06
        resultStream.min("amount").print("min");

        // minby操作,输出的信息都对
        resultStream.minBy("amount").print("minBy");

        resultStream.print("输出结果");

        // 触发程序执行
        env.execute("aggregation stream");
    }
}
